/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.store.index;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.rocketmq.common.ServiceThread;
import com.alibaba.rocketmq.common.UtilAll;
import com.alibaba.rocketmq.common.constant.LoggerName;
import com.alibaba.rocketmq.common.message.MessageConst;
import com.alibaba.rocketmq.common.sysflag.MessageSysFlag;
import com.alibaba.rocketmq.store.DefaultMessageStore;
import com.alibaba.rocketmq.store.DispatchRequest;
import com.alibaba.rocketmq.store.config.StorePathConfigHelper;

/**
 * 消息索引服务
 * ==========================================IndexService是线程类服务，在启动Broker时启动该线程服务。=========================================
 * 该类主要有两个功能，
 * 第一，是定时的创建消息的索引；
 * 第二是为应用层提供访问index索引文件的接口
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-7-21
 */
public class IndexService extends ServiceThread {
	private static final Logger log = LoggerFactory.getLogger(LoggerName.StoreLoggerName);
	private final DefaultMessageStore defaultMessageStore;
	// 索引配置
	private final int hashSlotNum;
	private final int indexNum;
	private final String storePath;
	// 索引文件集合
	private final ArrayList<IndexFile> indexFileList = new ArrayList<IndexFile>();
	// 读写锁（针对indexFileList）
	private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
	private LinkedBlockingQueue<Object[]> requestQueue = new LinkedBlockingQueue<Object[]>(300000);

	public IndexService(final DefaultMessageStore store) {
		this.defaultMessageStore = store;
		this.hashSlotNum = store.getMessageStoreConfig().getMaxHashSlotNum();
		this.indexNum = store.getMessageStoreConfig().getMaxIndexNum();
		this.storePath = StorePathConfigHelper.getStorePathIndex(store.getMessageStoreConfig().getStorePathRootDir());
	}

	public boolean load(final boolean lastExitOK) {
		// 加载$HOME/store/index目录下的文件
		File dir = new File(this.storePath);
		File[] files = dir.listFiles();
		if (files != null) {
			// ascending order
			Arrays.sort(files);
			for (File file : files) {
				try {
					// 对$HOME/store/index目录下的每个文件初始化一个IndexFile对象
					IndexFile f = new IndexFile(file.getPath(), this.hashSlotNum, this.indexNum, 0, 0);
					// 将IndexHeader加载到对象的变量中
					f.load();

					// 检查是否存在abort文件
					// 若存在abort文件，则表示Broker上次是异常退出的
					if (!lastExitOK) {
						if (f.getEndTimestamp() > this.defaultMessageStore.getStoreCheckpoint().getIndexMsgTimestamp()) { // this.defaultMessageStore.getStoreCheckpoint().getIndexMsgTimestamp()值表示最后刷盘的时间
							// 若小于则表示在最后刷盘之后在该文件中还创建了索引，则要删除该Index文件
							f.destroy(0);
							continue;
						}
					}
					// Broker上次是正常退出的
					// 或者Broker上次是异常退出的但在最后刷盘之后在该文件中没有创建索引
					log.info("load index file OK, " + f.getFileName());
					this.indexFileList.add(f);
				} catch (IOException e) {
					log.error("load file " + file + " error", e);
					return false;
				}
			}
		}

		return true;
	}

	/**
	 * 删除索引文件
	 */
	public void deleteExpiredFile(long offset) {
		Object[] files = null;
		try {
			this.readWriteLock.readLock().lock();
			if (this.indexFileList.isEmpty()) {
				return;
			}

			long endPhyOffset = this.indexFileList.get(0).getEndPhyOffset();
			if (endPhyOffset < offset) {
				files = this.indexFileList.toArray();
			}
		} catch (Exception e) {
			log.error("destroy exception", e);
		} finally {
			this.readWriteLock.readLock().unlock();
		}

		if (files != null) {
			List<IndexFile> fileList = new ArrayList<IndexFile>();
			for (int i = 0; i < (files.length - 1); i++) {
				IndexFile f = (IndexFile) files[i];
				if (f.getEndPhyOffset() < offset) {
					fileList.add(f);
				} else {
					break;
				}
			}

			this.deleteExpiredFile(fileList);
		}
	}

	/**
	 * 删除文件只能从头开始删
	 */
	private void deleteExpiredFile(List<IndexFile> files) {
		if (!files.isEmpty()) {
			try {
				this.readWriteLock.writeLock().lock();
				for (IndexFile file : files) {
					boolean destroyed = file.destroy(3000);
					destroyed = destroyed && this.indexFileList.remove(file);
					if (!destroyed) {
						log.error("deleteExpiredFile remove failed.");
						break;
					}
				}
			} catch (Exception e) {
				log.error("deleteExpiredFile has exception.", e);
			} finally {
				this.readWriteLock.writeLock().unlock();
			}
		}
	}

	public void destroy() {
		try {
			this.readWriteLock.readLock().lock();
			for (IndexFile f : this.indexFileList) {
				f.destroy(1000 * 3);
			}
			this.indexFileList.clear();
		} catch (Exception e) {
			log.error("destroy exception", e);
		} finally {
			this.readWriteLock.readLock().unlock();
		}
	}

	// ===========================查找topic和key的物理偏移量offset===========================
	public QueryOffsetResult queryOffset(String topic, String key, int maxNum, long begin, long end) {
		List<Long> phyOffsets = new ArrayList<Long>(maxNum);
		// TODO 可能需要返回给最终用户
		long indexLastUpdateTimestamp = 0;
		long indexLastUpdatePhyoffset = 0;
		maxNum = Math.min(maxNum, this.defaultMessageStore.getMessageStoreConfig().getMaxMsgsNumBatch());
		try {
			this.readWriteLock.readLock().lock();
			if (!this.indexFileList.isEmpty()) {
				for (int i = this.indexFileList.size(); i > 0; i--) {
					// 从IndexFile列表的最后一个对象开始往前遍历每个IndexFile对象
					IndexFile f = this.indexFileList.get(i - 1);
					boolean lastFile = (i == this.indexFileList.size());
					if (lastFile) {
						indexLastUpdateTimestamp = f.getEndTimestamp();
						indexLastUpdatePhyoffset = f.getEndPhyOffset();
					}
					// 检查开始时间、结束时间是否有一部分落在该Index文件中IndexHeader的beginTimestamp和EndTimestamp之间
					if (f.isTimeMatched(begin, end)) {
						// 最后一个文件需要加锁
						// 若是则调用IndexFile的selectPhyOffset方法获取物理偏移量列表
						f.selectPhyOffset(phyOffsets, this.buildKey(topic, key), maxNum, begin, end, lastFile);
					}

					// 再往前遍历时间更不符合
					// 检查IndexFile的beginTimestamp是否小于入参begin，
					if (f.getBeginTimestamp() < begin) {
						// 若是则不用在往前寻找了，直接返回
						break;
					}
					// 检查找到的物理偏移量个数是否已经大于64
					if (phyOffsets.size() >= maxNum) {
						// 若是则不用在往前寻找了，直接返回
						break;
					}
				}
			}
		} catch (Exception e) {
			log.error("queryMsg exception", e);
		} finally {
			this.readWriteLock.readLock().unlock();
		}

		return new QueryOffsetResult(phyOffsets, indexLastUpdateTimestamp, indexLastUpdatePhyoffset);
	}

	private String buildKey(final String topic, final String key) {
		return topic + "#" + key;
	}

	/**
	 * 向队列中添加请求，队列满情况下，丢弃请求
	 */
	public void putRequest(final Object[] reqs) {
		boolean offer = this.requestQueue.offer(reqs);
		if (!offer) {
			if (log.isDebugEnabled()) {
				log.debug("putRequest index failed, {}", reqs);
			}
		}
	}

	@Override
	public void run() {
		log.info(this.getServiceName() + " service started");

		while (!this.isStoped()) {
			try {
				Object[] req = this.requestQueue.poll(3000, TimeUnit.MILLISECONDS);

				if (req != null) {
					this.buildIndex(req);
				}
			} catch (Exception e) {
				log.warn(this.getServiceName() + " service has exception. ", e);
			}
		}

		log.info(this.getServiceName() + " service end");
	}

	/**
	 * ===============================创建消息的索引========================================
	 * @param req
	 */
	public void buildIndex(Object[] req) {
		boolean breakdown = false;
		// 获取Index文件的对象IndexFile
		IndexFile indexFile = retryGetAndCreateIndexFile();
		if (indexFile != null) {
			long endPhyOffset = indexFile.getEndPhyOffset();
			// 遍历requestQueue队列中的请求消息
			MSG_WHILE : for (Object o : req) {
				DispatchRequest msg = (DispatchRequest) o;
				String topic = msg.getTopic();
				String keys = msg.getKeys();
				// 将每个请求消息的commitlogOffset值与获取的IndexFile文件的endPhyOffset进行比较
				if (msg.getCommitLogOffset() < endPhyOffset) {
					// 若小于endPhyOffset值，则直接忽略该条请求信息
					continue;
				}

				final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
				switch (tranType) {
					case MessageSysFlag.TransactionNotType :
					case MessageSysFlag.TransactionPreparedType :
						break;
					case MessageSysFlag.TransactionCommitType :
					case MessageSysFlag.TransactionRollbackType :
						// 对于消息类型为Prepared和RollBack的也直接忽略掉。
						continue;
				}
				// 对于一个topic可以有多个key值
				if (keys != null && keys.length() > 0) {
					// 每个key值以空格分隔，
					String[] keyset = keys.split(MessageConst.KEY_SEPARATOR);
					// 遍历每个key值
					for (String key : keyset) {
						// TODO 是否需要TRIM
						if (key.length() > 0) {
							// 将topic-key值作为putKey方法的入参key值，将该topic的物理偏移量存入Index文件中，若存入失败则再次获取IndexFile对象重复调用putKey方法。
							for (boolean ok = indexFile.putKey(buildKey(topic, key), msg.getCommitLogOffset(), msg.getStoreTimestamp()); !ok;) {
								log.warn("index file full, so create another one, " + indexFile.getFileName());
								indexFile = retryGetAndCreateIndexFile();
								if (null == indexFile) {
									breakdown = true;
									break MSG_WHILE;
								}

								ok = indexFile.putKey(buildKey(topic, key), msg.getCommitLogOffset(), msg.getStoreTimestamp());
							}
						}
					}
				}
			}
		}
		// IO发生故障，build索引过程中断，需要人工参与处理
		else {
			breakdown = true;
		}

		if (breakdown) {
			log.error("build index error, stop building index");
		}
	}

	public IndexFile retryGetAndCreateIndexFile() {
		IndexFile indexFile = null;

		// 如果创建失败，尝试重建3次
		for (int times = 0; null == indexFile && times < 3; times++) {
			// 从IndexFile列表中获取最后一个IndexFile对象；若该对象对应的Index文件没有写满，即IndexHeader的indexCount不大于2000W；则直接返回该对象
			indexFile = this.getAndCreateLastIndexFile();
			if (null != indexFile)
				break;

			try {
				log.error("try to create index file, " + times + " times");
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

		// 重试多次，仍然无法创建索引文件
		if (null == indexFile) {
			this.defaultMessageStore.getAccessRights().makeIndexFileError();
			log.error("mark index file can not build flag");
		}

		return indexFile;
	}

	/**
	 * 获取最后一个索引文件，如果集合为空或者最后一个文件写满了，则新建一个文件<br>
	 * 只有一个线程调用，所以不存在写竟争问题
	 */
	public IndexFile getAndCreateLastIndexFile() {
		IndexFile indexFile = null;
		IndexFile prevIndexFile = null;
		long lastUpdateEndPhyOffset = 0;
		long lastUpdateIndexTimestamp = 0;
		// 先尝试使用读锁
		{
			this.readWriteLock.readLock().lock();
			if (!this.indexFileList.isEmpty()) {
				// 从IndexFile列表中获取最后一个IndexFile对象
				IndexFile tmp = this.indexFileList.get(this.indexFileList.size() - 1);
				// ；若该对象对应的Index文件没有写满，即IndexHeader的indexCount不大于2000W；则直接返回该对象
				if (!tmp.isWriteFull()) {
					indexFile = tmp;
				} else {
					// 若是因为写满了而创建，则在创建新Index文件时将该写满的Index文件的endPhyOffset和endTimestamp值初始化给新Index文件中IndexHeader的beginPhyOffset和beginTimestamp。
					lastUpdateEndPhyOffset = tmp.getEndPhyOffset();
					lastUpdateIndexTimestamp = tmp.getEndTimestamp();
					prevIndexFile = tmp;
				}
			}

			this.readWriteLock.readLock().unlock();
		}

		// 如果没找到，使用写锁创建文件
		// 若获得的该对象为空或者已经写满，则创建新的IndexFile对象，即新的Index文件
		// 若是因为写满了而创建，则在创建新Index文件时将该写满的Index文件的endPhyOffset和endTimestamp值初始化给新Index文件中IndexHeader的beginPhyOffset和beginTimestamp。
		if (indexFile == null) {
			try {
				String fileName = this.storePath + File.separator + UtilAll.timeMillisToHumanString(System.currentTimeMillis());
				indexFile = new IndexFile(fileName, this.hashSlotNum, this.indexNum, lastUpdateEndPhyOffset, lastUpdateIndexTimestamp);
				this.readWriteLock.writeLock().lock();
				this.indexFileList.add(indexFile);
			} catch (Exception e) {
				log.error("getLastIndexFile exception ", e);
			} finally {
				this.readWriteLock.writeLock().unlock();
			}

			// 每创建一个新文件，之前文件要刷盘
			// 启一个线程，调用IndexFile对象的fush将上一个写满的Index文件持久化到磁盘物理文件中
			if (indexFile != null) {
				final IndexFile flushThisFile = prevIndexFile;
				Thread flushThread = new Thread(new Runnable() {
					@Override
					public void run() {
						IndexService.this.flush(flushThisFile);
					}
				}, "FlushIndexFileThread");

				flushThread.setDaemon(true);
				flushThread.start();
			}
		}

		return indexFile;
	}

	public void flush(final IndexFile f) {
		if (null == f)
			return;

		long indexMsgTimestamp = 0;

		if (f.isWriteFull()) {
			indexMsgTimestamp = f.getEndTimestamp();
		}

		f.flush();

		if (indexMsgTimestamp > 0) {
			// 更新StoreCheckpoint.IndexMsgTimestamp为该写满的Index文件中IndexHeader的endTimestamp；
			this.defaultMessageStore.getStoreCheckpoint().setIndexMsgTimestamp(indexMsgTimestamp);
			this.defaultMessageStore.getStoreCheckpoint().flush();
		}
	}

	@Override
	public String getServiceName() {
		return IndexService.class.getSimpleName();
	}
}
